package com.wjk.kylin.rpc.core.netty.server;

import com.wjk.kylin.rpc.core.netty.params.RpcRequest;
import com.wjk.kylin.rpc.core.netty.params.RpcResponse;
import com.wjk.kylin.rpc.core.provider.RpcProviderFactory;
import com.wjk.kylin.rpc.core.util.ThrowableUtil;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ThreadPoolExecutor;


/**
 *
 * @author wjk
 */
@ChannelHandler.Sharable
public class NettyServerHandle extends SimpleChannelInboundHandler<RpcRequest> {
    private static final Logger LOGGER = LoggerFactory.getLogger(NettyServerHandle.class);
    private final RpcProviderFactory providerFactory;
    private final ThreadPoolExecutor serverHandlerPool;


    public NettyServerHandle(RpcProviderFactory providerFactory, ThreadPoolExecutor serverHandlerPool) {
        this.providerFactory = providerFactory;
        this.serverHandlerPool = serverHandlerPool;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcRequest request) throws Exception {
        LOGGER.debug(">>>>>>>>>>> kylin-rpc provider netty server read request:{}", request);

        // do invoke
        try {
            serverHandlerPool.execute(() -> {
                // invoke + response
                RpcResponse response = providerFactory.invokeService(request);

                LOGGER.debug(">>>>>>>>>>> kylin-rpc provider netty server write response {}", response);

                ctx.writeAndFlush(response);
            });
        } catch (Exception e) {
            // catch error
            RpcResponse rpcResponse = new RpcResponse();
            rpcResponse.setRequestId(request.getRequestId());
            rpcResponse.setErrorMsg(ThrowableUtil.toString(e));

            ctx.writeAndFlush(rpcResponse);
        }

    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        LOGGER.error(">>>>>>>>>>> kylin-rpc provider netty server caught exception", cause);
        ctx.close();
    }

}
